//! # 配置模块
//! 
//! 定义流处理相关的配置结构和管理功能

use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::time::Duration;
use crate::error::{StreamError, StreamResult};

/// 流处理配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamConfig {
    /// 流名称
    pub name: String,
    /// 输入配置
    pub input: InputConfig,
    /// 输出配置
    pub output: OutputConfig,
    /// 处理器配置列表
    pub processors: Vec<ProcessorConfig>,
    /// 错误处理配置
    pub error_handling: ErrorHandlingConfig,
    /// 检查点配置
    pub checkpointing: CheckpointConfig,
    /// 性能配置
    pub performance: PerformanceConfig,
    /// 监控配置
    pub monitoring: MonitoringConfig,
}

impl StreamConfig {
    /// 创建新的流配置
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            input: InputConfig::default(),
            output: OutputConfig::default(),
            processors: Vec::new(),
            error_handling: ErrorHandlingConfig::default(),
            checkpointing: CheckpointConfig::default(),
            performance: PerformanceConfig::default(),
            monitoring: MonitoringConfig::default(),
        }
    }

    /// 验证配置
    pub fn validate(&self) -> StreamResult<()> {
        if self.name.is_empty() {
            return Err(StreamError::config_error("流名称不能为空"));
        }

        self.input.validate()?;
        self.output.validate()?;
        self.error_handling.validate()?;
        self.checkpointing.validate()?;
        self.performance.validate()?;

        Ok(())
    }

    /// 添加处理器配置
    pub fn add_processor(mut self, processor: ProcessorConfig) -> Self {
        self.processors.push(processor);
        self
    }

    /// 设置错误处理配置
    pub fn with_error_handling(mut self, error_handling: ErrorHandlingConfig) -> Self {
        self.error_handling = error_handling;
        self
    }

    /// 设置检查点配置
    pub fn with_checkpointing(mut self, checkpointing: CheckpointConfig) -> Self {
        self.checkpointing = checkpointing;
        self
    }
}

/// 输入配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InputConfig {
    /// 输入类型
    pub input_type: InputType,
    /// 主题模式
    pub topic_pattern: String,
    /// 消费者组
    pub consumer_group: Option<String>,
    /// 并发度
    pub concurrency: u32,
    /// 缓冲区大小
    pub buffer_size: usize,
    /// 超时设置
    pub timeout_ms: u64,
    /// 自定义属性
    pub properties: HashMap<String, String>,
}

impl Default for InputConfig {
    fn default() -> Self {
        Self {
            input_type: InputType::MessageBus,
            topic_pattern: String::new(),
            consumer_group: None,
            concurrency: 1,
            buffer_size: 1000,
            timeout_ms: 5000,
            properties: HashMap::new(),
        }
    }
}

impl InputConfig {
    /// 验证输入配置
    pub fn validate(&self) -> StreamResult<()> {
        if self.topic_pattern.is_empty() {
            return Err(StreamError::config_error("主题模式不能为空"));
        }

        if self.concurrency == 0 {
            return Err(StreamError::config_error("并发度必须大于0"));
        }

        if self.buffer_size == 0 {
            return Err(StreamError::config_error("缓冲区大小必须大于0"));
        }

        Ok(())
    }
}

/// 输入类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum InputType {
    /// 消息总线
    MessageBus,
    /// 文件
    File { path: String },
    /// HTTP端点
    Http { endpoint: String },
    /// 自定义
    Custom { handler: String },
}

/// 输出配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OutputConfig {
    /// 输出类型
    pub output_type: OutputType,
    /// 目标主题
    pub target_topic: String,
    /// 批处理大小
    pub batch_size: usize,
    /// 刷新间隔
    pub flush_interval_ms: u64,
    /// 确认模式
    pub ack_mode: AckMode,
    /// 自定义属性
    pub properties: HashMap<String, String>,
}

impl Default for OutputConfig {
    fn default() -> Self {
        Self {
            output_type: OutputType::MessageBus,
            target_topic: String::new(),
            batch_size: 100,
            flush_interval_ms: 1000,
            ack_mode: AckMode::Auto,
            properties: HashMap::new(),
        }
    }
}

impl OutputConfig {
    /// 验证输出配置
    pub fn validate(&self) -> StreamResult<()> {
        if self.target_topic.is_empty() {
            return Err(StreamError::config_error("目标主题不能为空"));
        }

        if self.batch_size == 0 {
            return Err(StreamError::config_error("批处理大小必须大于0"));
        }

        Ok(())
    }
}

/// 输出类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum OutputType {
    /// 消息总线
    MessageBus,
    /// 文件
    File { path: String },
    /// HTTP端点
    Http { endpoint: String },
    /// 数据库
    Database { connection_string: String },
    /// 自定义
    Custom { handler: String },
}

/// 确认模式
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum AckMode {
    /// 自动确认
    Auto,
    /// 手动确认
    Manual,
    /// 批量确认
    Batch,
}

/// 处理器配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProcessorConfig {
    /// 处理器名称
    pub name: String,
    /// 处理器类型
    pub processor_type: String,
    /// 处理器参数
    pub parameters: HashMap<String, serde_json::Value>,
    /// 是否启用
    pub enabled: bool,
    /// 处理器顺序
    pub order: i32,
}

impl ProcessorConfig {
    /// 创建新的处理器配置
    pub fn new(name: impl Into<String>, processor_type: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            processor_type: processor_type.into(),
            parameters: HashMap::new(),
            enabled: true,
            order: 0,
        }
    }

    /// 添加参数
    pub fn with_parameter(mut self, key: impl Into<String>, value: serde_json::Value) -> Self {
        self.parameters.insert(key.into(), value);
        self
    }

    /// 设置顺序
    pub fn with_order(mut self, order: i32) -> Self {
        self.order = order;
        self
    }
}

/// 错误处理配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ErrorHandlingConfig {
    /// 重试配置
    pub retry: RetryConfig,
    /// 死信队列配置
    pub dead_letter: DeadLetterConfig,
    /// 错误策略
    pub strategy: ErrorStrategy,
    /// 错误阈值
    pub error_threshold: f64,
    /// 熔断配置
    pub circuit_breaker: Option<CircuitBreakerConfig>,
}

impl Default for ErrorHandlingConfig {
    fn default() -> Self {
        Self {
            retry: RetryConfig::default(),
            dead_letter: DeadLetterConfig::default(),
            strategy: ErrorStrategy::Retry,
            error_threshold: 0.1, // 10% 错误率
            circuit_breaker: None,
        }
    }
}

impl ErrorHandlingConfig {
    /// 验证错误处理配置
    pub fn validate(&self) -> StreamResult<()> {
        if self.error_threshold < 0.0 || self.error_threshold > 1.0 {
            return Err(StreamError::config_error("错误阈值必须在0.0-1.0之间"));
        }

        self.retry.validate()?;
        self.dead_letter.validate()?;

        Ok(())
    }
}

/// 重试配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RetryConfig {
    /// 最大重试次数
    pub max_attempts: u32,
    /// 重试间隔策略
    pub backoff_strategy: BackoffStrategy,
    /// 最大重试间隔
    pub max_delay_ms: u64,
    /// 可重试的错误类型
    pub retryable_errors: Vec<String>,
}

impl Default for RetryConfig {
    fn default() -> Self {
        Self {
            max_attempts: 3,
            backoff_strategy: BackoffStrategy::Exponential,
            max_delay_ms: 30000, // 30秒
            retryable_errors: vec![
                "TIMEOUT_ERROR".to_string(),
                "NETWORK_ERROR".to_string(),
                "RESOURCE_EXHAUSTED".to_string(),
            ],
        }
    }
}

impl RetryConfig {
    /// 验证重试配置
    pub fn validate(&self) -> StreamResult<()> {
        if self.max_attempts == 0 {
            return Err(StreamError::config_error("最大重试次数必须大于0"));
        }

        if self.max_delay_ms == 0 {
            return Err(StreamError::config_error("最大重试间隔必须大于0"));
        }

        Ok(())
    }

    /// 计算重试延迟
    pub fn calculate_delay(&self, attempt: u32) -> Duration {
        let delay_ms = match self.backoff_strategy {
            BackoffStrategy::Fixed => 1000, // 1秒
            BackoffStrategy::Linear => 1000 * attempt as u64,
            BackoffStrategy::Exponential => {
                let base_delay = 1000u64; // 1秒
                let exponential_delay = base_delay * 2u64.pow(attempt.saturating_sub(1));
                exponential_delay.min(self.max_delay_ms)
            }
        };

        Duration::from_millis(delay_ms.min(self.max_delay_ms))
    }
}

/// 退避策略
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum BackoffStrategy {
    /// 固定间隔
    Fixed,
    /// 线性增长
    Linear,
    /// 指数增长
    Exponential,
}

/// 死信队列配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeadLetterConfig {
    /// 是否启用
    pub enabled: bool,
    /// 死信主题
    pub topic: String,
    /// 最大重试次数
    pub max_retries: u32,
    /// 保留时间（小时）
    pub retention_hours: u32,
    /// 是否包含原始头信息
    pub include_original_headers: bool,
}

impl Default for DeadLetterConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            topic: "dead-letter-queue".to_string(),
            max_retries: 3,
            retention_hours: 24,
            include_original_headers: true,
        }
    }
}

impl DeadLetterConfig {
    /// 验证死信队列配置
    pub fn validate(&self) -> StreamResult<()> {
        if self.enabled && self.topic.is_empty() {
            return Err(StreamError::config_error("启用死信队列时主题不能为空"));
        }

        Ok(())
    }
}

/// 错误策略
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ErrorStrategy {
    /// 重试
    Retry,
    /// 死信队列
    DeadLetter,
    /// 跳过
    Skip,
    /// 中止
    Abort,
}

/// 熔断器配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CircuitBreakerConfig {
    /// 失败率阈值
    pub failure_rate_threshold: f64,
    /// 最小请求数阈值
    pub minimum_number_of_calls: u32,
    /// 等待时间（毫秒）
    pub wait_duration_ms: u64,
    /// 半开状态最大调用数
    pub permitted_number_of_calls_in_half_open_state: u32,
}

impl Default for CircuitBreakerConfig {
    fn default() -> Self {
        Self {
            failure_rate_threshold: 0.5, // 50%
            minimum_number_of_calls: 10,
            wait_duration_ms: 60000, // 1分钟
            permitted_number_of_calls_in_half_open_state: 3,
        }
    }
}

/// 检查点配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CheckpointConfig {
    /// 是否启用检查点
    pub enabled: bool,
    /// 检查点间隔（毫秒）
    pub interval_ms: u64,
    /// 检查点存储类型
    pub storage_type: CheckpointStorageType,
    /// 存储路径
    pub storage_path: String,
    /// 是否自动保存
    pub auto_save: bool,
}

impl Default for CheckpointConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            interval_ms: 10000, // 10秒
            storage_type: CheckpointStorageType::File,
            storage_path: "./checkpoints".to_string(),
            auto_save: true,
        }
    }
}

impl CheckpointConfig {
    /// 验证检查点配置
    pub fn validate(&self) -> StreamResult<()> {
        if self.enabled {
            if self.interval_ms == 0 {
                return Err(StreamError::config_error("检查点间隔必须大于0"));
            }

            if self.storage_path.is_empty() {
                return Err(StreamError::config_error("存储路径不能为空"));
            }
        }

        Ok(())
    }
}

/// 检查点存储类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum CheckpointStorageType {
    /// 文件存储
    File,
    /// 数据库存储
    Database,
    /// Redis存储
    Redis,
    /// 内存存储（仅用于测试）
    Memory,
}

/// 性能配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PerformanceConfig {
    /// 最大并发度
    pub max_concurrency: u32,
    /// 背压策略
    pub backpressure_strategy: BackpressureStrategy,
    /// 缓冲区大小
    pub buffer_size: usize,
    /// 批处理配置
    pub batch_config: BatchConfig,
    /// 内存限制（MB）
    pub memory_limit_mb: Option<u64>,
}

impl Default for PerformanceConfig {
    fn default() -> Self {
        Self {
            max_concurrency: num_cpus::get() as u32,
            backpressure_strategy: BackpressureStrategy::Block,
            buffer_size: 10000,
            batch_config: BatchConfig::default(),
            memory_limit_mb: None,
        }
    }
}

impl PerformanceConfig {
    /// 验证性能配置
    pub fn validate(&self) -> StreamResult<()> {
        if self.max_concurrency == 0 {
            return Err(StreamError::config_error("最大并发度必须大于0"));
        }

        if self.buffer_size == 0 {
            return Err(StreamError::config_error("缓冲区大小必须大于0"));
        }

        self.batch_config.validate()?;

        Ok(())
    }
}

/// 背压策略
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum BackpressureStrategy {
    /// 阻塞
    Block,
    /// 丢弃最旧的
    DropOldest,
    /// 丢弃最新的
    DropNewest,
    /// 抛出错误
    Error,
}

/// 批处理配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BatchConfig {
    /// 启用批处理
    pub enabled: bool,
    /// 批大小
    pub size: usize,
    /// 批超时（毫秒）
    pub timeout_ms: u64,
}

impl Default for BatchConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            size: 100,
            timeout_ms: 1000,
        }
    }
}

impl BatchConfig {
    /// 验证批处理配置
    pub fn validate(&self) -> StreamResult<()> {
        if self.enabled {
            if self.size == 0 {
                return Err(StreamError::config_error("批大小必须大于0"));
            }

            if self.timeout_ms == 0 {
                return Err(StreamError::config_error("批超时必须大于0"));
            }
        }

        Ok(())
    }
}

/// 监控配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MonitoringConfig {
    /// 启用指标收集
    pub metrics_enabled: bool,
    /// 指标端口
    pub metrics_port: Option<u16>,
    /// 启用健康检查
    pub health_check_enabled: bool,
    /// 健康检查端口
    pub health_check_port: Option<u16>,
    /// 日志级别
    pub log_level: String,
    /// 追踪采样率
    pub trace_sampling_rate: f64,
}

impl Default for MonitoringConfig {
    fn default() -> Self {
        Self {
            metrics_enabled: true,
            metrics_port: Some(9090),
            health_check_enabled: true,
            health_check_port: Some(8080),
            log_level: "info".to_string(),
            trace_sampling_rate: 0.1, // 10%
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_stream_config_validation() {
        let mut config = StreamConfig::new("test-stream");
        config.input.topic_pattern = "test.*".to_string();
        config.output.target_topic = "output".to_string();
        
        assert!(config.validate().is_ok());
    }

    #[test]
    fn test_input_config_validation() {
        let mut config = InputConfig::default();
        config.topic_pattern = "test.*".to_string();
        assert!(config.validate().is_ok());

        config.topic_pattern = String::new();
        assert!(config.validate().is_err());
    }

    #[test]
    fn test_retry_config_delay_calculation() {
        let config = RetryConfig::default();
        
        let delay1 = config.calculate_delay(1);
        let delay2 = config.calculate_delay(2);
        let delay3 = config.calculate_delay(3);
        
        // 指数退避：1s, 2s, 4s
        assert_eq!(delay1, Duration::from_millis(1000));
        assert_eq!(delay2, Duration::from_millis(2000));
        assert_eq!(delay3, Duration::from_millis(4000));
    }

    #[test]
    fn test_processor_config() {
        let config = ProcessorConfig::new("filter", "filter")
            .with_parameter("condition", serde_json::json!("age > 18"))
            .with_order(1);
        
        assert_eq!(config.name, "filter");
        assert_eq!(config.processor_type, "filter");
        assert_eq!(config.order, 1);
        assert!(config.parameters.contains_key("condition"));
    }
}